package com.flute.haflute.agent;

import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.manage.JvmId;
import com.flute.haflute.agent.manage.JvmManager;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.CLI;
import com.flute.haflute.common.ClusterConstants.COMMAND;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.master.Master;

/**
 * 一项工作执行的主控节点——工作控制节点/中心（ControlCenter）
 * 
 * 一项工作的调度和执行模式如下:<br>
 * 
 * 1. 一项工作分为多个任务，每个任务由一个子JVM执行，每一个SLAVE节点管理一个或者多个子JVM。<br>
 * 2. 任务执行中，每一个子JVM向ControlCenter进行数据通信，同时向SLAVE进行状态汇报，以完成监控的功能
 */
public class JobControlCenterManager {
	private static Logger logger = LoggerFactory
			.getLogger(JobControlCenterManager.class);

	public void startControllJvm() {
		File libPath = new File("./lib");
		JvmManager jvmManager = JvmManager.getInstance(Master.class.getName(),
				libPath);

		// should check if has child, if has shutdown first. actually currently,
		// jobtracker will exit when task finished
		if (jvmManager.getAllProcess().size() > 0) {
			syncJvmStatusWitchManager(jvmManager);
			// shutdownAllChildJvms(jvmManager);

		}

		if (jvmManager.getAllProcess().size() == 0)
			startAllChildJvms(jvmManager);
	}

	private void syncJvmStatusWitchManager(JvmManager jvmManager) {
		Map<JvmId, Process> processes = jvmManager.processMap();
		for (Iterator<JvmId> it = processes.keySet().iterator(); it.hasNext();) {
			JvmId id = it.next();
			Process proc = jvmManager.getProcess(id);
			try {
				int exitCode = proc.exitValue();
				logger.info("one process exit code={}", exitCode);
				it.remove();
			} catch (IllegalThreadStateException e) {
			}
		}
	}

	private void startAllChildJvms(JvmManager jvmManager) {
		// 启动 master.Master
		File workDirFile = new File("master");
		if (!workDirFile.exists()) {
			if (!workDirFile.mkdirs()) {
				System.err.println("can not Make workdir: master/");
			}
		}

		ArrayList<String> argList = new ArrayList<String>();
		argList.add(CLI.ROOTDIR_IS + ClusterContext.getRootDir());
		String[] jvmParams = new String[0];
		if (ClusterContext.isJvmDebugging()) {
			jvmParams = new String[2];
			jvmParams[0] = "-Xdebug";
			jvmParams[1] = "-Xrunjdwp:transport=dt_socket,address=6666,server=y,suspend=n";
		}

		try {
			Process process = jvmManager.getProcess(jvmManager.spawnChildJvm(
					workDirFile, jvmParams,
					argList.toArray(new String[argList.size()])));
			BufferedReader reader = new BufferedReader(new InputStreamReader(
					process.getInputStream()));
			String line;
			while ((line = reader.readLine()) != null) {
				// 重要! 该项控制必须一致
				if (ClusterConstants.PROCESS_STARTED.trim().equals(line)) {
					break;
				}
				System.out.println(line);
			}
			reader.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void shutdownAllChildJvms(JvmManager jvmManager) {
		Collection<Process> processes = jvmManager.getAllProcess();
		for (Process process : processes) {
			JvmManager.sendCommand2OneJvm(process, COMMAND.SHUTDOWN);
			logger.info("wait for Master process to finish");
			try {
				process.waitFor();
			} catch (InterruptedException e) {
			}
			logger.info("Master process Finished");
		}
		jvmManager.clearProcessCache();
	}

	public static void main(String args[]) {
		Runtime.getRuntime().addShutdownHook(new ShutdownHook());
		JobControlCenterManager a = new JobControlCenterManager();
		a.startControllJvm();
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.exit(0);
	}
}
